package kafka.examples;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

public class ConsumerDemo {
    public static void main(String[] args) {
        String topicName = "test-topic";
        String groupId = "test-group";

        Properties props = new Properties();
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.SttringDeserializer");
        // 这里可以配置几台broker即可，他会自动从broker去拉取元数据进行缓存
        props.put("bootstrap.servers", "hadoop03:9092,hadoop04:9092,hadoop05:9092");
        props.put("group.id", groupId);

        /**
         * consumer自动提交offset, 默认为true。每个一段时间自动提交。
         * 如果一旦重启消费者，必会出现重复消费问题。
         */
        props.put("enable.auto.commit", "true");
        //consumer自动提交offset的时间间隔
        props.put("auto.commit.ineterval.ms", "1000");

        /**
         * earliest: 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
         * latest: 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
         * none: topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
         */
        props.put("auto.offset.reset", "earliest");


        //获取一条消息最大的字节数, 建议10MB
        props.put("fetch.max.bytes",10485760);
        //一次poll返回消息的最大条数，默认是500条。如果业务消费的吞吐量特别大，可以适当提高一点。
        props.put("max.poll.records",500);
        //consumer与broker的socket的连接如果空闲时间超过了这个时间，就会自动回收。下次消费要重新建立连接。-1：不要去回收。
        props.put("connection.max.idle.ms",-1);


        /**
         * consumer心跳时间，这个时间尽可能的短点，
         * 如果说某个comsumer挂了，kafka broker感知到了，会触发一个rebalance操作，就是重新分配分区给其他的consumer,
         * 其他的consumer如果感知到了 rebalance重新分配分区，就会通过心跳来感知
         * 心跳的间隔一般不要设置太长，1000，500
         */
        props.put("heartbeat.interval.ms",100);
        /**
         * 如果kafka broker 在10秒内感知不到一个consumer的心跳，就会认为它挂了。默认10秒。
         * 会触发rebalance操作。
         */
        props.put("session.timeout.ms",10 * 1000);
        /**
         * 如果在两次poll操作之间，超过了这个时间，那么就会认为这个consume处理能力太弱了，会被踢出消费组，分区分配给别人去消费
         * 一般来结合自己的业务处理的性能来设置，
         */
        props.put("max.poll.interval.ms",20*1000);

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        consumer.subscribe(Arrays.asList(topicName));

        try {
            while(true) {
                ConsumerRecords<String, String> records = consumer.poll(1000); // 超时时间
                for(ConsumerRecord<String, String> record : records) {
                    System.out.println(record.offset() + ", " + record.key() + ", " + record.value());
                }
            }
        } catch(Exception e) {

        }
    }
}
